#ifndef ACE_UDP_CONNECTOR_IMPL_H
#define ACE_UDP_CONNECTOR_IMPL_H

#include "ace/SOCK_Dgram.h"
#include "ace/Svc_Handler.h"
#include "ace/Reactor.h"
#include "ace/TP_Reactor.h"
#include "../udp_message_context.h"
#include "../udp_connector_config.h"
#include "../udp_session.h"
#include "../io_handler.h"
#include <string>
#include <boost/bind.hpp>

namespace vpm
{
	namespace mina
	{
		namespace detail
		{		
			class AceUdpConnectorImpl  : public  ACE_Svc_Handler<ACE_SOCK_Dgram,ACE_MT_SYNCH> 
			{
				enum
				{
					UDP_MAX_RECV_BUF_LEN = 8192,
				};

			public:
				AceUdpConnectorImpl(UdpConnectorConfig& config ,IoHandler& handler):m_udpConfig(config),m_ioHandler(handler),m_thrdGrpId(-1)
				{
					m_pReactor = new ACE_Reactor( new ACE_TP_Reactor(),1); 					

					m_udpSession.reset(new UdpSession());
					IoSessionAccessor::attach_writter(*m_udpSession,boost::bind(&AceUdpConnectorImpl::send_data,this,_1));
					IoSessionAccessor::attach_reader(*m_udpSession,boost::bind(&AceUdpConnectorImpl::recv_data,this,_1));					
				}

				virtual ~AceUdpConnectorImpl()
				{
					this->shutdown(); 
					this->reactor(NULL);
					delete m_pReactor; 
				}

				int connect(int port =0, const std::string & host="0.0.0.0");

				void disconnect();

				virtual ACE_HANDLE get_handle (void) const
				{
					return m_dgramSock.get_handle();
				}

				int svc();	

				int handle_input(ACE_HANDLE handle);

				int handle_output (ACE_HANDLE fd )
				{
					return 0; 
				}

				int handle_close (ACE_HANDLE handle, ACE_Reactor_Mask close_mask)
				{
					m_dgramSock.close(); 
					return 0; 
				}

				int process_data(char * pData, int dataLen,int peerPort, const std::string & peerHost)
				{
					IoFilterChain & chain = m_udpConfig.get_input_chain();
					UdpMessageContext context(pData,dataLen,peerPort,peerHost); 		
					chain.process(context);		
					return 0; 
				}

				int send_data(MessageContext & msg)
				{
					UdpMessageContext & udpMsg = (UdpMessageContext&) msg;
					return this->send_data(udpMsg.buffer, udpMsg.length,udpMsg.port,udpMsg.get_host_addr());		
				}

				int send_data(const char * pData, size_t dataLen, int port, const std::string & host)
				{		
					return m_dgramSock.send(pData,dataLen,ACE_INET_Addr(port,host.c_str()));  	
				}

				int recv_data(MessageContext & msg)
				{					
					UdpMessageContext & udpMsg = (UdpMessageContext&) msg;
					ACE_INET_Addr peerAddr;
					int ret = m_dgramSock.recv((void*)udpMsg.buffer,(size_t)udpMsg.length,peerAddr);
					if (ret >=0)
					{
						udpMsg.port = peerAddr.get_port_number();
						udpMsg.ip   = peerAddr.get_ip_address();
					}
					return ret; 
				}			

			private:
				ACE_SOCK_Dgram m_dgramSock; 
				ACE_Reactor * m_pReactor; 				

				UdpConnectorConfig& m_udpConfig; 
				IoHandler & m_ioHandler;
				int m_thrdGrpId; 
				std::auto_ptr<IoSession> m_udpSession;
			};

#define  UdpConnectorImpl AceUdpConnectorImpl
		}
	}
}
#endif // 
  
